import json
from typing import List, Optional

from redis.asyncio import Redis
from sqlalchemy import delete, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.core.config import Settings
from app.exceptions.base import AppException
from app.models.system import SysRole, SysRoleMenu
from app.schemas.base import Pagination
from app.schemas.role import RoleCreate, RoleForm, RolePageVO


class RoleService:
    """角色服务类"""

    def __init__(self, session: AsyncSession, redis: Redis, settings: Settings):
        """初始化角色服务
        Args:
            session: 数据库会话
        """
        self.session = session
        self.redis = redis
        self.settings = settings

    async def get_role_by_id(self, role_id: int) -> Optional[SysRole]:
        """
        根据角色ID获取角色信息
        Args:
            role_id: 角色ID
        Returns:
            Optional[SysRole]: 角色信息
        """
        stmt = (
            select(SysRole).where(SysRole.id == role_id, SysRole.is_deleted == 0).options(selectinload(SysRole.menus))
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def get_role_by_code(self, codes: List[str] = None) -> Optional[List[SysRole]]:
        """
        根据角色编码获取角色信息
        Args:
            code: 角色编码
        Returns:
            Optional[List[SysRole]]: 角色信息
        """
        stmt = select(SysRole).where(SysRole.is_deleted == 0, SysRole.status == 1)
        if codes:
            stmt = stmt.where(SysRole.code.in_(codes))

        stmt = stmt.options(selectinload(SysRole.menus))

        result = await self.session.execute(stmt)
        return result.scalars().all()

    async def get_role_page(
        self, keywords: str, is_root: bool, page_num: int = 1, page_size: int = 10
    ) -> Pagination[RolePageVO]:
        """
        获取角色列表
        Args:
            role_name: 角色名称
            status: 角色状态
        Returns:
            List[SysRole]: 角色列表
        """
        stmt = select(SysRole).where(SysRole.is_deleted == 0)

        if keywords:
            stmt = stmt.where(
                or_(
                    SysRole.name.ilike(func.concat("%", keywords, "%")),
                    SysRole.code.ilike(func.concat("%", keywords, "%")),
                )
            )

        # 非超级管理员不显示超级管理员角色
        if not is_root:
            stmt = stmt.where(SysRole.code != self.settings.ADMIN_ROLE_CODE)

        total_query = select(func.count()).select_from(stmt.subquery())
        result = await self.session.execute(total_query)
        total = result.scalar()

        stmt = stmt.order_by(SysRole.sort.asc(), SysRole.create_time.desc(), SysRole.update_time.desc())

        offset = (page_num - 1) * page_size

        paginated_query = stmt.offset(offset).limit(page_size)
        result = await self.session.execute(paginated_query)
        items = result.scalars().all()

        return Pagination(
            list=[RolePageVO.model_validate(role) for role in items],
            total=total,
            page_num=page_num,
            page_size=page_size,
        )

    async def refresh_role_perms_cache(self, role_code: str = None, old_role_code: str = None) -> None:
        """
        优化后的权限缓存刷新
        """
        # 1. 缓存键清理优化
        if not role_code and not old_role_code:
            # 全量刷新：直接删除整个哈希表
            await self.redis.delete(self.settings.KEY_SYSTEM_ROLE_PERMS)
        else:
            # 批量删除指定键
            keys_to_delete = []
            if role_code:
                keys_to_delete.append(role_code)
            if old_role_code:
                keys_to_delete.append(old_role_code)

            if keys_to_delete:
                await self.redis.hdel(self.settings.KEY_SYSTEM_ROLE_PERMS, *keys_to_delete)

        # 2. 智能查询优化
        if not role_code and not old_role_code:
            roles = await self.get_role_by_code()
        elif role_code:
            roles = await self.get_role_by_code([role_code])
        else:
            # 仅old_role_code场景，无需查询权限
            return

        # 3. 批量更新缓存（使用管道优化）
        async with self.redis.pipeline() as pipe:
            for role in roles:
                perms = {menu.perm for menu in role.menus if menu.perm}
                perms_json = json.dumps(list(perms))
                pipe.hset(self.settings.KEY_SYSTEM_ROLE_PERMS, role.code, perms_json)
            await pipe.execute()

    async def get_role_list(self, role_ids: List[int] = None) -> Optional[List[SysRole]]:
        """
        根据角色ID获取角色信息
        Args:
            role_id: 角色ID
        Returns:
            Optional[SysRole]: 角色信息
        """
        stmt = select(SysRole).where(SysRole.is_deleted == 0)
        if role_ids:
            stmt = stmt.where(SysRole.id.in_(role_ids))
        result = await self.session.execute(stmt)
        return result.scalars().all()

    async def save_role(self, role_id: int, role_form: RoleForm, user_id: int = None) -> SysRole:
        """
        保存角色（新增或编辑）
        Args:
            role_id: 角色ID（为空时新增，非空时编辑）
            role_form: 角色表单数据
            user_id: 当前用户ID（可选）
        Returns:
            SysRole: 保存后的角色信息
        """
        old_role = None
        role = None
        old_code = None
        old_status = None

        # 1. 角色存在性检查（修复：添加逻辑删除检查）
        if role_id:
            result = await self.session.execute(select(SysRole).where(SysRole.id == role_id, SysRole.is_deleted == 0))
            old_role = result.scalar_one_or_none()
            if not old_role:
                raise AppException(message="角色不存在")
            # 记录变更前状态
            old_code = old_role.code
            old_status = old_role.status

        # 2. 唯一性验证（保持不变）
        result = await self.session.execute(
            func.count(SysRole)
            .where(SysRole.id != role_id)
            .where(SysRole.is_deleted == 0)  # 修复：排除已删除角色
            .where(
                or_(
                    SysRole.name == role_form.name,
                    SysRole.code == role_form.code,
                )
            )
        )
        count = result.scalar()
        if count > 0:
            raise AppException(message="角色名称或角色编码已存在，请修改后重试！")

        # 3. 保存逻辑（保持不变）
        if old_role:
            ext_data = {"update_time": func.now()}
            if user_id:
                ext_data["update_by"] = user_id
            old_role.sqlmodel_update(role_form.model_dump(exclude_unset=True), update=ext_data)
            self.session.add(old_role)
            role = old_role
        else:
            role_create = RoleCreate.model_validate(role_form)
            role = SysRole.model_validate(role_create)
            self.session.add(role)

        # 检查关键字段变更（修复：新增角色也需要刷新）
        code_changed = old_code and old_code != role.code
        status_changed = old_status is not None and old_status != role.status
        is_new_role = not old_role  # 新增角色标识

        # 4. 权限缓存刷新（修复：正确的条件判断）
        if code_changed or status_changed or is_new_role:
            if code_changed and old_code:
                # 角色编码变更：需要同时处理新旧编码
                await self.refresh_role_perms_cache(role_code=role.code, old_role_code=old_code)
            elif is_new_role:
                # 新增角色：刷新所有缓存
                await self.refresh_role_perms_cache()
            else:
                # 角色信息变更：只刷新当前角色
                await self.refresh_role_perms_cache(role_code=role.code)

            # 提交数据库事务（确保缓存刷新成功后再提交）
            await self.session.commit()
            await self.session.flush()

        return role

    async def delete_role(self, role_id: int, user_id: int = None) -> None:
        """
        删除角色
        Args:
            role_id: 角色ID
        Returns:
            None
        """
        role = await self.get_role_by_id(role_id)
        if not role:
            raise AppException(message="角色不存在")
        role.is_deleted = 1
        role.update_time = func.now()  # 修复：更新更新时间
        if user_id:
            role.update_by = user_id

        self.session.add(role)
        await self.session.commit()
        await self.session.flush()
        # 刷新权限缓存
        await self.refresh_role_perms_cache(role_code=None, old_role_code=role.code)

    async def assign_role_menus(self, role_id: int, menu_ids: List[int]) -> None:
        """
        分配角色菜单权限
        Args:
            role_id: 角色ID
            menu_ids: 菜单ID列表
        Returns:
            None
        Raises:
            AppException: 角色不存在时抛出异常
        """
        # 1. 查询角色信息
        role = await self.get_role_by_id(role_id)

        # 2. 验证角色是否存在
        if not role:
            raise AppException(message="角色不存在")

        # 3. 删除角色原有菜单权限
        delete_stmt = delete(SysRoleMenu).where(SysRoleMenu.role_id == role_id)
        await self.session.execute(delete_stmt)
        # result.rowcount  # 确保删除操作生效

        # 4. 条件判断：menuIds是否为空
        if menu_ids:  # 批量添加新权限
            self.session.add_all([SysRoleMenu(role_id=role_id, menu_id=menu_id) for menu_id in menu_ids])

        # 5. 刷新角色权限缓存
        await self.refresh_role_perms_cache(role_code=role.code)

        # 提交事务
        await self.session.commit()
        await self.session.flush()
